package com.ctbri.vSearch.frame;


import java.net.InetAddress;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.log4j.Logger;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.xpack.client.PreBuiltXPackTransportClient;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Repository;

import com.ctbri.vSearch.util.PropUtil;

/**
 * ES数据连接池
 * 
 * @author Hogan
 * 
 */
@Repository("eSDataPool")
@Scope("singleton")
public class ESDataPool {
	
	private final int TRYTIMES = 5;
	private Logger log = Logger.getLogger(getClass());
	private AtomicInteger contActive;
	/**
	 * 空闲连接队列
	 */
	public ArrayBlockingQueue<TransportClient> maxConnection ;
	/**
	 * 活动连接队列
	 */
	private ConcurrentHashMap<TransportClient,Integer> activeConnection ;
	/**
	 * 将线程和连接绑定，保证事务能统一执行
	 */
	private ThreadLocal<TransportClient> threadLocal = new ThreadLocal<TransportClient>();;
	private Lock releaselock = new ReentrantLock();
	private ESDataBean eSDataBean;
	private static int count=0;
	private ESDataPool(){
		System.out.println("初始化ES连接！"+(++count)+"次！！！");
	}
	//初始化信息
	@PostConstruct
	public void startPool() {
		this.eSDataBean = loadESDataBean();
		this.contActive = new AtomicInteger(0); 
		this.activeConnection = new ConcurrentHashMap<>();
		this.threadLocal = new ThreadLocal<TransportClient>();
		this.maxConnection = new ArrayBlockingQueue<>(eSDataBean.maxActiveConnections);
		initPool();
	}
	
	public TransportClient getCurrentConnection(){
		TransportClient tran =  threadLocal.get();
		if(tran == null){
			tran = getConnection();
		}
		return tran;
	}
	
	public boolean releaseConnection(TransportClient trans){
		if(!activeConnection.containsKey(trans)){
			return false;
		}
		releaselock.lock();
		try {
			
			threadLocal.remove();
			activeConnection.remove(trans);
			maxConnection.put(trans);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}finally{
			releaselock.unlock();
		}
		return true;
	}
	
	/***
	 * 移除一个连接
	 */
	public boolean destroyConnection(TransportClient trans){
		if(trans == null){
			return true;
		}
		if(!activeConnection.containsKey(trans)){
			return false;
		}
		releaselock.lock();
		try{
			threadLocal.remove();
			activeConnection.remove(trans);
			trans.close();
		}catch (Exception e) {
			e.printStackTrace();
		}finally{
			releaselock.unlock();
		}
		return true;
	}
	
	/**
	 * 销毁线程池
	 */
//	@PreDestroy
	public void shutdownPool(){
		
		System.out.println("关闭ES连接池！");
		Set<TransportClient> keys = activeConnection.keySet();
		Iterator<TransportClient> ikeys =keys.iterator();
		while(ikeys.hasNext()){
			TransportClient tc = ikeys.next();
			activeConnection.remove(tc);
			tc.close();
		}
		while(!maxConnection.isEmpty()){
			TransportClient tc = maxConnection.poll();
			tc.close();
		}
	}
	
	 /**
     * 获得连接
     * 1. 判断是否有空闲连接如果有返回
     * 2. 如果没有，查看是否可以创建新连接放入maxConnection中,如果可以创建
     * 3. 如果不可以 ,阻塞等待
     * @return
     */
	private TransportClient getConnection() {
		Lock putactivelock = new ReentrantLock();
    	AtomicInteger errTimes = new AtomicInteger(TRYTIMES);
		TransportClient trans =null;
		try{
			while(trans == null && errTimes.get()>0){
				if(maxConnection.size()>0){
					trans =maxConnection.poll();
					if(trans != null){
						threadLocal.set(trans);
						activeConnection.put(trans, 0);
					}else{
						 errTimes.decrementAndGet();
		            	 System.out.println(Thread.currentThread().getName()+"没有空闲线程"+errTimes.get());
					}
				}else{
					 //如果活动线程数小于活动队列可存放空间
		             trans = newConnection();//可能为空
		             if(trans != null ){
		            	 putactivelock.lock();
		            	 threadLocal.set(trans);
		            	 activeConnection.put(trans,0);//添加当前活动链接
		            	 putactivelock.unlock();
		             }else{
		            	 errTimes.decrementAndGet();
		             }
				}
			}
		} catch (Exception e) {  
            log.error("获取连接失败!",e);
            trans = null;
        } 
		return trans;
	}
	
	private void initPool(){
		AtomicInteger errNum =new AtomicInteger(TRYTIMES);
    	while(contActive.get()<eSDataBean.getInitConnections()) { 
    		TransportClient conn = newConnection();
            // 初始化最小连接数  
            if (conn == null){
    			if(errNum.decrementAndGet()<0){
    				log.info("初始化连接容忍失败次数："+errNum.get());
    				break;
    			}
            }else {
                try {
					maxConnection.put(conn);
					contActive.incrementAndGet();//记录当前存活的线程  
				} catch (InterruptedException e) {
					e.printStackTrace();
					System.out.println("初始化中断异常");
				}
                System.out.println("成功加入："+conn.hashCode()+" "+contActive.get()+" initnums"+eSDataBean.getInitConnections());
            }
            
        }  
    	log.info("初始化ES连接数为："+ contActive);
	}
	
	@SuppressWarnings("resource")
	private TransportClient newConnection() {
		TransportClient client = null;
		System.out.println("准备开启新连接");
		Settings settings = Settings.builder()
				.put("cluster.name", eSDataBean.getCluserName())
				.put("xpack.security.user", eSDataBean.getUsername())
				.build();
		try {
			client = new PreBuiltXPackTransportClient(settings)
					.addTransportAddress(new InetSocketTransportAddress(
							InetAddress.getByName(eSDataBean.getIp()),eSDataBean.getPort()));
		} catch (Exception e) {
			log.error("获取连接失败!",e);
			client = null;
		}
		return client;
	}
	
	private ESDataBean loadESDataBean(){
		ESDataBean eSDataBean= new ESDataBean();
		eSDataBean.setCluserName(PropUtil.getValue("es.cluster.name"));
		eSDataBean.setUsername(PropUtil.getValue("es.xpack.security.user"));
		eSDataBean.setIp(PropUtil.getValue("es.ip"));
		eSDataBean.setPort(Integer.valueOf(PropUtil.getValue("es.port")));
		eSDataBean.setPoolName(PropUtil.getValue("es.poolName"));
		eSDataBean.setMinConnections(Integer.valueOf(PropUtil.getValue("es.minConnections")));
		eSDataBean.setMaxConnections(Integer.valueOf(PropUtil.getValue("es.maxConnections")));
		eSDataBean.setInitConnections(Integer.valueOf(PropUtil.getValue("es.initConnections")));
		eSDataBean.setConnTimeOut(Long.valueOf(PropUtil.getValue("es.connTimeOut")));
		eSDataBean.setMaxActiveConnections(Integer.valueOf(PropUtil.getValue("es.maxActiveConnections")));
		eSDataBean.setConnectionTimeOut(Integer.valueOf(PropUtil.getValue("es.connectionTimeOut")));
		return eSDataBean;
	}
	
	private static class ESDataBean {
		/**
		 * 集群名称
		 */
		private String cluserName;
		/**
		 * 用户名
		 */
		private String username;
		/**
		 * 连接的IP地址
		 */
		private String ip;
		/**
		 * 连接的IP端口
		 */
		private int port;
		/**
		 * 连接池名称
		 */
		private String poolName;
		/**
		 * 最小连接数
		 */
		private int minConnections;
		/**
		 * 最大连接数
		 */
		private int maxConnections;
		/**
		 * 初始化连接数
		 */
		private int initConnections;
		/**
		 * 重复获得连接的频率
		 */
		private long connTimeOut;
		/**
		 * 最大允许的连接数，和数据库对应
		 */
		private int maxActiveConnections;
		/**
		 * 连接超时时间
		 */
		private long connectionTimeOut;
		
		public String getCluserName() {
			return cluserName;
		}
		public void setCluserName(String cluserName) {
			this.cluserName = cluserName;
		}
		public String getUsername() {
			return username;
		}
		public void setUsername(String username) {
			this.username = username;
		}
		public String getIp() {
			return ip;
		}
		public void setIp(String ip) {
			this.ip = ip;
		}
		public int getPort() {
			return port;
		}
		public void setPort(int port) {
			this.port = port;
		}
		public String getPoolName() {
			return poolName;
		}
		public void setPoolName(String poolName) {
			this.poolName = poolName;
		}
		public int getMinConnections() {
			return minConnections;
		}
		public void setMinConnections(int minConnections) {
			this.minConnections = minConnections;
		}
		public int getMaxConnections() {
			return maxConnections;
		}
		public void setMaxConnections(int maxConnections) {
			this.maxConnections = maxConnections;
		}
		public int getInitConnections() {
			return initConnections;
		}
		public void setInitConnections(int initConnections) {
			this.initConnections = initConnections;
		}
		public long getConnTimeOut() {
			return connTimeOut;
		}
		public void setConnTimeOut(long connTimeOut) {
			this.connTimeOut = connTimeOut;
		}
		public int getMaxActiveConnections() {
			return maxActiveConnections;
		}
		public void setMaxActiveConnections(int maxActiveConnections) {
			this.maxActiveConnections = maxActiveConnections;
		}
		public long getConnectionTimeOut() {
			return connectionTimeOut;
		}
		public void setConnectionTimeOut(long connectionTimeOut) {
			this.connectionTimeOut = connectionTimeOut;
		}
	}
}
